package com.atguigu.func;

import com.atguigu.beans.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * @author Felix
 * @date 2024/4/3
 */
public class MySinkFunction extends RichSinkFunction<WaterSensor> {
    private Connection conn;
    @Override
    public void open(Configuration parameters) throws Exception {
        //注册驱动
        Class.forName("com.mysql.cj.jdbc.Driver");
        //建立连接
        conn = DriverManager.getConnection(
                "jdbc:mysql://hadoop102:3306/test?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8",
                "root",
                "123456"
        );
    }

    @Override
    public void close() throws Exception {
        if(conn != null){
            conn.close();
        }
    }

    @Override
    public void invoke(WaterSensor ws, Context context) throws Exception {
        //创建数据库操作对象
        PreparedStatement ps = conn.prepareStatement("insert into ws values(?,?,?)");
        ps.setString(1, ws.id);
        ps.setLong(2, ws.getTs());
        ps.setString(3, ws.getVc() + "");
        //执行SQL
        ps.execute();
        //释放资源
        if(ps != null){
            ps.close();
        }
    }
}
